package org.egomsl.mw.region;

import org.apache.hadoop.fs.*;
import org.egomsl.mw.record.Record;
import org.egomsl.mw.record.RecordScanner;
import org.egomsl.mw.segment.SegmentRecordScanner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;

public class RegionReader {
    private static Logger log = LoggerFactory.getLogger(RegionReader.class);

    protected FileSystem fs;
    protected Path regionDir;

    protected List<FileMeta> storefiles;

    public RegionReader(FileSystem fs, String dir) throws IOException {
        this(fs, new Path(dir));
    }

    public RegionReader(FileSystem fs, Path dir) throws IOException {
        this.fs = fs;
        this.regionDir = dir;
        this.storefiles = RegionUtils.getSegmentFiles(fs, dir);
    }

    public RegionReader(FileSystem fs, Path dir, List<FileMeta> storefiles) {
        this.fs = fs;
        this.regionDir = dir;
        this.storefiles = storefiles;
    }

    public int getStoreFileCount() {
        return storefiles.size();
    }

    public FileMeta getStoreFile(int i) {
        return storefiles.get(i);
    }

    public RecordScanner getScanner(boolean filtered) {
        RegionRecordScanner regionRecordScanner =new RegionRecordScanner(this);
        if(filtered) {
            return new FilteredRegionRecordScanner(regionRecordScanner);
        }else {
            return regionRecordScanner;
        }
    }

    public static class RegionRecordScanner implements RecordScanner {

        private int nextStoreFileIndex;
        private SegmentRecordScanner segmentRecordScanner;
        private RegionReader regionReader;
        public RegionRecordScanner(RegionReader regionReader) {
            nextStoreFileIndex = 0;
            segmentRecordScanner = null;
            this.regionReader = regionReader;
        }

        @Override
        public boolean hashNext() {
            try {
                if(segmentRecordScanner != null && segmentRecordScanner.hashNext()) {
                    return true;
                }else if(nextStoreFileIndex < regionReader.getStoreFileCount()){
                    if(segmentRecordScanner != null) segmentRecordScanner.close();

                    FileMeta fileMeta = regionReader.getStoreFile(nextStoreFileIndex++);

                    segmentRecordScanner = new SegmentRecordScanner(this.regionReader.fs, fileMeta.getPath());
                    if(segmentRecordScanner.hashNext()) return true;
                }
            } catch (IOException e) {
                log.error("Error: ", e);
            }
            return false;
        }

        @Override
        public Record next() {
            if(segmentRecordScanner != null)
                return segmentRecordScanner.next();
            return null;
        }
    }

    public static class FilteredRegionRecordScanner implements RecordScanner {
        protected RecordScanner rawRecordScanner;
        protected Set<Record.Key> tagedRecords;
        protected Record nextRecord;

        public FilteredRegionRecordScanner(RecordScanner recordScanner) {
            this.rawRecordScanner = recordScanner;
            tagedRecords = new HashSet<>();
            nextRecord = null;
        }

        @Override
        public boolean hashNext() {
            if(rawRecordScanner == null) return false;

            while(rawRecordScanner.hashNext()) {
                nextRecord = rawRecordScanner.next();
                Record.Key key = nextRecord.getKey();

                switch (key.getTag()) {
                    case DELETE:
                        tagedRecords.add((Record.Key)key.clone());
                        nextRecord = null;
                        break;
                    case INSERT:
                        if(!tagedRecords.contains(key)) {
                            return true;
                        }
                        break;
                    case UPDATE:
                        if(!tagedRecords.contains(key)) {
                            tagedRecords.add((Record.Key)key.clone());
                            return true;
                        }
                        break;
                        default:
                            throw new RuntimeException("Unexpected tag.");
                }
            }

            return false;
        }

        @Override
        public Record next() {
            return nextRecord;
        }
    }

}
